package com.inspur.cloud.controller.resource;

import com.inspur.cloud.entity.dataspace.UserEntity;
import com.inspur.cloud.entity.dataspace.ao.resource.DistributeAO;
import com.inspur.cloud.entity.dataspace.common.OpEntityResult;
import com.inspur.cloud.entity.dataspace.common.PageReqParam;
import com.inspur.cloud.entity.dataspace.common.PageResult;
import com.inspur.cloud.entity.dataspace.vo.KafkaAddVO;
import com.inspur.cloud.entity.dataspace.vo.KafkaReq;
import com.inspur.cloud.entity.dataspace.vo.KafkaVO;
import com.inspur.cloud.security.AuthorizationHelper;
import com.inspur.cloud.service.dataspace.KafkaManagerService;
import com.inspur.cloud.util.StringUtil;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import javax.annotation.security.RolesAllowed;

@RestController
@RequestMapping("/dataspace/api/v1/kafka")
@Api(value = "kafka", description = "kafka资源管理")
public class KafkaManagerController {
    @Autowired
    KafkaManagerService kafkaManagerService;

    @RequestMapping(value = "/list", method = {RequestMethod.POST})
    @ApiOperation(value = "kafka资源列表", notes = "")
    public PageResult<KafkaVO> showList(@ApiParam(value = "kafka查询对象,json格式请求")
                                        @RequestBody PageReqParam<KafkaReq> kafkaReq) {
        PageResult<KafkaVO> result = new PageResult<>(1, "查询成功");
        if (kafkaReq.getReqParam() == null) {
            result.setCode(-1);
            result.setMsg("参数不完整");
            return result;
        }
        if (kafkaReq.getPageSize() > 50 || kafkaReq.getPageSize() < 0) {
            result.setCode(-1);
            result.setMsg("页码非法");
            return result;
        }
        String resourceName=kafkaReq.getReqParam().getResourceName();
        if(!StringUtil.isEmpty(resourceName)){
            kafkaReq.getReqParam().setResourceName(StringUtil.specialStr(resourceName));
        }
        result=kafkaManagerService.showUserList(kafkaReq);
        return result;
    }


    @RequestMapping(value = "/addRes", method = {RequestMethod.POST})
    @ApiOperation(value = "新建kafka资源", notes = "")
    @RolesAllowed({"Admin", "Tenant"})
    public OpEntityResult<Integer> addRes(@ApiParam(value = "kafka新增对象")
                                          @RequestBody KafkaAddVO kafkaVO) throws Exception {
        UserEntity user=AuthorizationHelper.getUser();
        if (StringUtil.isEmpty(kafkaVO.getTopicName())) {
            return OpEntityResult.fail("TOPIC 名不能为空");
        }
        if (!StringUtil.isLetterAndNumber(kafkaVO.getTopicName())) {
            return OpEntityResult.fail("只能输入数字和字母");
        }
        if (kafkaVO.getPartitionNum() == null || kafkaVO.getPartitionNum() <= 0) {
            return OpEntityResult.fail("分区数输入不合法");
        }
        if (kafkaVO.getReplication() == null || kafkaVO.getReplication() <= 0) {
            return OpEntityResult.fail("副本数输入不合法");
        }
        if(kafkaVO.getType()==null){
            kafkaVO.setType(1);//默认创建公共资源
        }
        return kafkaManagerService.addRes(kafkaVO,user);
    }

    @RequestMapping(value = "/modifyRes", method = {RequestMethod.POST})
    @ApiOperation(value = "编辑kafka资源", notes = "注意：仅支持增加分区数目，不能减少。副本数也不支持编辑")
    @RolesAllowed({"Admin", "Tenant"})
    public OpEntityResult<Integer> modifyRes(@ApiParam(value = "kafka编辑对象")
                                             @RequestBody KafkaAddVO kafkaVO) {
        if (kafkaVO.getResourceId() == null) {
            return OpEntityResult.fail("资源ID不可为空");
        }
        if (kafkaVO.getPartitionNum() == null || kafkaVO.getPartitionNum() <= 0) {
            return OpEntityResult.fail("分区数输入不合法");
        }
        return kafkaManagerService.modifyRes(kafkaVO);
    }

    @RequestMapping(value = "/delRes", method = {RequestMethod.DELETE})
    @ApiOperation(value = "删除kafka资源", notes = "只有管理员可以删除")
    @RolesAllowed({"Admin"})
    public OpEntityResult<Integer> delRes(@ApiParam(value = "资源ID")
                                          @RequestParam(value = "resourceId", required = true) Integer resourceId
    ) {
        if (resourceId == null) {
            return OpEntityResult.fail("资源ID不可为空");
        }
        return kafkaManagerService.delRes(resourceId);
    }

    @PostMapping("/distributeRes")
    @ApiOperation(value = "kafka分配资源", notes = "")
    @RolesAllowed({"Admin", "Tenant"})
    public OpEntityResult<Integer> distributeTenant(@ApiParam(value = "资源对象")
                                                    @RequestBody DistributeAO kafkaVO) throws Exception {
        if (kafkaVO.getResourceId() == null) {
            return OpEntityResult.fail("资源ID不可为空");
        }
        OpEntityResult<Integer> result = kafkaManagerService.distributeRes(kafkaVO);
        return result;
    }

    @PostMapping("/shareRes")
    @ApiOperation(value = "kafka分享资源", notes = "")
    @RolesAllowed({"Admin", "Tenant"})
    public OpEntityResult<Integer> shareTenant(@ApiParam(value = "资源对象")
                                               @RequestBody DistributeAO kafkaVO) throws Exception {
        if (kafkaVO.getResourceId() == null) {
            return OpEntityResult.fail("资源ID不可为空");
        }
        OpEntityResult<Integer> result = kafkaManagerService.share(kafkaVO);
        return result;
    }

    @GetMapping("/sync")
    @ApiOperation(value = "同步kafka信息", notes = "")
    @RolesAllowed({"Admin", "Tenant"})
    public OpEntityResult<Integer> sync() {
        OpEntityResult<Integer> result = kafkaManagerService.sync();
        return result;
    }

    @GetMapping("/getReplicationNum")
    @ApiOperation(value = "获取KAFKA支持的最大副本数", notes = "副本数不能超过broken数")
    @RolesAllowed({"Admin", "Tenant"})
    public OpEntityResult<Integer> getReplicationNum() {
        OpEntityResult<Integer> result = new OpEntityResult<>();
        result.setCode(1);
        result.setData(kafkaManagerService.getReplicationNum());
        result.setMsg("最大副本数获取成功");
        return result;
    }
}
